package com.gugee.ins.data.collector.media.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.beanutil.InsAppMediaConversion;
import com.gugee.ins.common.beanutil.InsAppProductConversion;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.common.dal.service.media.*;
import com.gugee.ins.common.dal.service.product.InsProductDao;
import com.gugee.ins.common.dal.spider.InsMediaSpiderClient;
import com.gugee.ins.common.model.mysql.blogger.InsBloggerMain;
import com.gugee.ins.common.model.es.product.InsProductEsMapping;
import com.gugee.ins.common.model.mysql.media.InsLocation;
import com.gugee.ins.common.model.mysql.media.InsMedia;
import com.gugee.ins.common.model.mysql.media.InsRelMediaTaggerUser;
import com.gugee.ins.common.model.mysql.product.InsProduct;
import com.gugee.ins.common.model.mysql.product.InsRelMediaProduct;
import com.gugee.ins.common.model.spider.app.InsAppBloggerNode;
import com.gugee.ins.common.model.spider.app.media.*;
import com.gugee.ins.common.model.spider.app.product.InsAppProductBody;
import com.gugee.ins.common.model.spider.app.product.InsAppProductTag;
import com.gugee.ins.data.collector.media.config.InsMediaConfig;
import com.gugee.ins.data.collector.media.config.InsProductMapping;
import com.gugee.ins.data.collector.media.service.InsMediaService;
import com.gugee.ins.data.collector.media.service.InsProductService;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import com.xiguaji.boot.toolkit.elasticsearch.RestClientHelper;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ForkJoinPool;
import java.util.regex.Pattern;

/**
 * insBlogger服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsMediaServiceImpl implements InsMediaService {

    /**
     * 配置
     */
    @Autowired
    InsMediaConfig insMediaConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * 爬虫接口
     */
    @Autowired
    InsMediaSpiderClient insMediaSpiderClient;

    /**
     * InsBloggerExtDao
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     * insBloggerMainDao
     */
    @Autowired
    InsBloggerMainDao insBloggerMainDao;


    /**
     * InsMediaDao
     */
    @Autowired
    InsMediaDao insMediaDao;

    /**
     *
     */
    @Autowired
    InsLocationDao insLocationDao;

    /**
     * InsRelMediaTaggerUserDao
     */
    @Autowired
    InsRelMediaTaggerUserDao insRelMediaTaggerUserDao;

    /**
     * InsRelMediaCaptionTagDao
     */
    @Autowired
    InsRelMediaCaptionTagDao insRelMediaCaptionTagDao;


    /**
     * InsRelMediaCaptionTaggerDao
     */
    @Autowired
    InsRelMediaCaptionTaggerDao insRelMediaCaptionTaggerDao;

    /**
     * insRelMediaSponsorUserDao
     */
    @Autowired
    InsRelMediaSponsorUserDao insRelMediaSponsorUserDao;

    /**
     * insProductDao
     */
    @Autowired
    InsProductDao insProductDao;


    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsBlogger;

    /**
     *
     */
    @Autowired
    InsProductService insProductService;

    @Autowired
    InsProductMapping insProductMapping;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    private String workName = "insMediaCollect-worker";

    /**
     * username 的 匹配串
     */
    private static final String USERNAME_PATTERN = "^(?!\\.)(?!.*\\.$)(?!.*?\\.\\.)[a-zA-Z0-9_.]+$";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(insMediaConfig.getInsMediaCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 处理insMedia收入任务
     * @param insBloggerMessages
     */
    @Override
    public void handleInsMediaCollect(List<InsBloggerMessage> insBloggerMessages) {
        try {
            forkJoinPool.submit(() -> insBloggerMessages.parallelStream().forEach(msg -> {
                try {
                    //采集播主关注人数据
                    runSpiderInsMedia(msg);
                } catch (Exception ex) {
                    log.error("InsMedia collect error. uid:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("InsMedia collect  thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 采集更新帖子数据
     * @param insBloggerMessage
     */
    public void runSpiderInsMedia(InsBloggerMessage insBloggerMessage){
        long start = System.currentTimeMillis();
        long uid = insBloggerMessage.getUid();
        // 根据uid爬取信息
        String pageInfo;
        // 粉丝数
        long followerCount;
        // 30天时间区间
        long timeRang = start / 1000 - insMediaConfig.getLatestDays() * 24 * 3600;
        val insBlogger = insBloggerMainDao.findByUid(uid);

        // 最近采集限制时间
        int limitDay = insMediaConfig.getInsMediaIntervalDay();
        if(insBloggerMessage.getTypeCode() == InsBloggerMsgTypeCodeEnum.UID_UPDATE_F_BLOGGER_TIMING_MIDDLE.getCode()){
            limitDay = insMediaConfig.getInsMiddleBloggerMediaIntervalDay();
        }
        long lastCollectTime = start / 1000 - limitDay * 24 * 3600;
        // 如果查的到 且 帖子数大于设置数 且 最近采集过 且 账户私有 那么才采集
        if(insBlogger != null && insBlogger.getMediaCount() > insMediaConfig.getInsMediaMinCount()
                && !insBlogger.getIsPrivate()
                && insBlogger.getLastMediaCollectTime() < lastCollectTime){
            // 获取粉丝数计算参与率
            followerCount = insBlogger.getFollowerCount();

            // 商品相关信息
            List<InsProduct> products = new ArrayList<>(10);
            List<InsRelMediaProduct> relMp = new ArrayList<>(10);
            Set<Long> productIds = new HashSet<>(10);

            pageInfo = spiderMedia(uid,"",followerCount,timeRang,
                    products,relMp,productIds);
            while(!StringUtils.isEmpty(pageInfo)){
                pageInfo = spiderMedia(uid,pageInfo,followerCount,timeRang,
                        products,relMp,productIds);
            }

            // 更新帖子平均er值
            try{
                insBloggerMainDao.updateAvgEr(uid,timeRang);
            }catch (Exception e){
                log.error("InsMedia updateAvgEr error, uid:{},msg:{}", uid,e.getMessage());
            }

            // 保存商品相关操作，同时调用标记商品接口获取标记商品的帖子
            if(!products.isEmpty()){
                saveProductsAndDoSpider(products,relMp,productIds);
            }

            // 发送消息
            sendMsg(insBloggerMessage,uid,insBlogger,start);

        }

        if ((System.currentTimeMillis() - start) / 1000 > insMediaConfig.getMediaDebugTimeLimit()) {
            log.info("InsMedia collect uid:{} use time:{}s", insBloggerMessage.getUid(),(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 循环拉帖子
     * @param uid
     * @param endCursor
     * @param followerCount
     * @param timeRang
     * @return
     */
    public String spiderMedia(long uid,String endCursor,long followerCount,long timeRang,
                              List<InsProduct> products, List<InsRelMediaProduct> relMp,Set<Long> productIds){
        //根据uid爬取信息
        List<InsAppMediaBody> edges;
        InsAppMediaResponse insAppMediaResponse;
        String pageInfo;
        boolean saveUpdateTime = false;

        try {
            insAppMediaResponse = insMediaSpiderClient.getAppMediaByUid(uid,endCursor);
            edges = insAppMediaResponse.getItems();
            if(edges != null && !edges.isEmpty()){
                for(InsAppMediaBody edge : edges){
                    if(edge != null){
                        if(edge.getTakenTime() > timeRang){
                            try{
                                //保存帖子及相关信息
                                saveMedia(edge,followerCount);

                                // 更新标记字段
                                if(!saveUpdateTime){
                                    insBloggerMainDao.updateMediaCollectTime(uid);
                                    saveUpdateTime = true;
                                }
                            }catch (Exception e){
                                log.error("media merge error,msg:{}",e.getMessage());
                            }

                            // 处理商品相关
                            getProduct(edge,uid,products,relMp,productIds);
                        }else{
                            //直接返回空，不再爬取
                            return null;
                        }
                    }
                }
            }
            pageInfo = insAppMediaResponse.getNextMaxId();
        }catch (Exception e){
            pageInfo = null;
            log.error("spider ins app media collect error, msg :{}",e.getMessage());
        }
        return pageInfo;
    }

    /**
     * 保存帖子及相关信息
     * @param edge
     * @param followerCount
     */
    @Override
    public void saveMedia(InsAppMediaBody edge,long followerCount){
        // 帖子
        InsMedia insMedia = InsAppMediaConversion.mapper(edge);

        // 计算参与率
        insMedia.setEr(followerCount <= 0 ? 0 : (double)(insMedia.getCommentCount() + insMedia.getLikeCount()) / followerCount);
        edge.setEr(insMedia.getEr());
        InsMedia insMediaDba = insMediaDao.findByShortcode(insMedia.getShortcode());
        long nowTime = System.currentTimeMillis() / 1000;
        // 判断帖子是否已经保存过，保存了后面的信息就不用处理了
        if(insMediaDba != null){
            // 更新帖子
            insMedia.setId(insMediaDba.getId());
            insMedia.setLastCollectTime(nowTime);
            insMediaDao.updateById(insMedia);
        }else{
            insMedia.setCreateTime(nowTime);
            insMedia.setLastCollectTime(nowTime);
            // 处理帖子标题
            if(!StringUtils.isEmpty(insMedia.getCaption()) ){
                saveMediaCaption(insMedia.getShortcode(),insMedia.getCaption(),insMedia.getTakenTime());
                if(insMedia.getCaption().length() > 4096){
                    insMedia.setCaption(insMedia.getCaption().substring(0,4095));
                }
            }

            // 获取额外信息
            Set<Long> sponsoredUid = new HashSet<>(10);
            Set<Long> taggedUid = new HashSet<>(10);
            getMediaTaggerAndSponsored(edge,sponsoredUid,taggedUid);

            if(!sponsoredUid.isEmpty()){
                insMedia.setIsSponsored(true);
            }
            // 保存帖子
            insMediaDao.save(insMedia);

            // 保存赞助人信息
            if(!sponsoredUid.isEmpty()){
                saveSponsored(insMedia.getShortcode(),insMedia.getTakenTime(),insMedia.getOwnerUid(),sponsoredUid);
            }
            // 保存标记人
            if(!taggedUid.isEmpty()){
                saveUserTagger(insMedia.getShortcode(),insMedia.getOwnerUid(),insMedia.getTakenTime(),new ArrayList<>(taggedUid),nowTime);
            }
            // 地理位置
            if(insMedia.getLocationId() != null){
                saveLocation(nowTime,insMedia);
            }
        }
    }

    /**
     * 保存帖子的标题提及内容
     * @param shortcode
     * @param caption
     */
    public void saveMediaCaption(String shortcode,String caption,long takenTime){
        Set<String> taggers = new HashSet<>(10);
        Set<String> tags = new HashSet<>(10);
        for(int i = 0; i < caption.length(); ++i){
            if('@' == caption.charAt(i)){
                i = getTagger(caption,i + 1,taggers);
            }
            if('#' == caption.charAt(i)){
                i = getTag(caption,i + 1,tags);
            }
        }
        if(!taggers.isEmpty()){
            insRelMediaCaptionTaggerDao.batchInsert(shortcode,takenTime, new ArrayList<>(taggers));
        }
        if(!tags.isEmpty()){
            insRelMediaCaptionTagDao.batchInsert(shortcode, new ArrayList<>(tags));
        }
    }

    /**
     * 获取提及话题
     * @param caption
     * @param index
     * @param tags
     * @return
     */
    private int getTag(String caption,int index,Set<String> tags){
        StringBuilder tag = new StringBuilder();
        for(int i = index; i < caption.length(); ++i){
            if(')' == caption.charAt(i) || ' ' == caption.charAt(i) || ',' == caption.charAt(i) || '.' == caption.charAt(i)
                    || '，' == caption.charAt(i) || '。' == caption.charAt(i) || '、' == caption.charAt(i)
                    || '!' == caption.charAt(i) || '?' == caption.charAt(i) || ':' == caption.charAt(i)
                    || '！' == caption.charAt(i) || '？' == caption.charAt(i) || '：' == caption.charAt(i)
                    || '\n' == caption.charAt(i) || '\t' == caption.charAt(i)){
                index = i;
                break;
            }else if('@' == caption.charAt(i) || '#' == caption.charAt(i)){
                index = i - 1;
                break;
            }else{
                tag.append(caption.charAt(i));
            }
        }
        if(tag.length() > 0 ){
            tags.add(tag.toString());
        }
        return index;
    }

    /**
     * 获取提及人
     * @param caption
     * @param index
     * @param tags
     * @return
     */
    private int getTagger(String caption,int index,Set<String> tags){
        StringBuilder tag = new StringBuilder();
        for(int i = index; i < caption.length(); ++i){
            if('@' == caption.charAt(i) || '#' == caption.charAt(i)){
                index = i - 1;
                break;
            }else{
                tag.append(caption.charAt(i));
                if(!Pattern.matches(USERNAME_PATTERN, tag)){
                    tag.deleteCharAt(tag.length() - 1);
                    break;
                }
            }
        }
        if(tag.length() > 0 ){
            tags.add(tag.toString());
        }
        return index;
    }

    /**
     * 发送采集评论的消息
     * @param insBloggerMessage
     */
    public void sendCollectCommentMsg(InsBloggerMessage insBloggerMessage){
        insBloggerMessage.setRetryLimitTimes(insMediaConfig.getInsCommentPriorityRetryTimes());
        insBloggerMessage.setRetryTimes(insMediaConfig.getInsCommentPriorityRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsBlogger.send(insMediaConfig.getInsCommentCollectPriorityTopic(),insBloggerMessage);
    }

    /**
     * 发送识别广告帖的消息
     * @param insBloggerMessage
     */
    public void sendIdentifyAdMediaMsg(InsBloggerMessage insBloggerMessage){
        insBloggerMessage.setRetryLimitTimes(insMediaConfig.getInsAdMediaIdentifyRetryTimes());
        insBloggerMessage.setRetryTimes(insMediaConfig.getInsAdMediaIdentifyRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsBlogger.send(insMediaConfig.getInsAdMediaIdentifyPriorityTopic(),insBloggerMessage);
    }

    /**
     * 发送分析视频分数
     * @param insBloggerMessage
     */
    public void sendAnalysisMediaScoreMsg(InsBloggerMessage insBloggerMessage){
        insBloggerMessage.setRetryLimitTimes(insMediaConfig.getInsAnalysisMediaScoreRetryTimes());
        insBloggerMessage.setRetryTimes(insMediaConfig.getInsAnalysisMediaScoreRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsBlogger.send(insMediaConfig.getInsAnalysisMediaScoreOnlineTopic(),insBloggerMessage);
    }

    /**
     * 发送分析播主分类消息
     * @param insBloggerMessage
     */
    public void sendAnalysisCategoryMsg(InsBloggerMessage insBloggerMessage){
        insBloggerMessage.setRetryLimitTimes(insMediaConfig.getInsAnalysisCategoryRetryTimes());
        insBloggerMessage.setRetryTimes(insMediaConfig.getInsAnalysisCategoryRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsBlogger.send(insMediaConfig.getInsAnalysisCategoryOnlineTopic(),insBloggerMessage);
    }

    /**
     * 获取赞助人信息
     * @param edges
     */
    public void getSponsored(List<InsAppMediaSponsorBody> edges,Set<Long> uids){
        if(edges != null && !edges.isEmpty()){
            InsAppBloggerNode sponsor;
            // 保存赞助商信息
            for(InsAppMediaSponsorBody sponsors : edges){
                try{
                    if(sponsors != null && sponsors.getSponsor() != null){
                        sponsor = sponsors.getSponsor();
                        if(sponsor.getUid() != null
                                && sponsor.getUid() > 0
                                && !uids.contains(sponsor.getUid())){
                            uids.add(sponsor.getUid());
                            sendPriorityCollectBloggerMsg(sponsor.getUid(),sponsor.getUsername(),
                                    sponsor.getPicUrl(),sponsor.getIsPrivate() != null ? sponsor.getIsPrivate() : false,
                                    InsBloggerMsgTypeCodeEnum.CLP_SRC_MEDIA_SPONSOR.getCode());

                        }
                    }
                }catch (Exception e){
                    log.error("get media Sponsored Error,msg:{}",e.getMessage());
                }
            }
        }
    }

    /**
     * 保存赞助人信息
     * @param shortcode
     * @param uid
     * @param sponsoredUid
     */
    public void saveSponsored(String shortcode,long takenTime,long uid,Set<Long> sponsoredUid){
        insRelMediaSponsorUserDao.batchInsert(shortcode,takenTime,uid,new ArrayList<>(sponsoredUid));
    }

    /**
     * 获取提及用户及赞助用户信息
     * @param edge
     * @param sponsoredUid
     * @param taggedUid
     */
    public void getMediaTaggerAndSponsored(InsAppMediaBody edge,Set<Long> sponsoredUid,Set<Long> taggedUid){
        getSponsored(edge.getSponsorTags(),sponsoredUid);
        if(edge.getUsertags() != null && edge.getUsertags().getIn() != null
                && !edge.getUsertags().getIn().isEmpty()){
            getUserTagger(edge.getUsertags().getIn(),taggedUid);
        }
        getChildrenMediaTaggerAndSponsored(edge.getCarouselMedia(),sponsoredUid,taggedUid);
    }

    /**
     * 获取子帖子中提及用户及赞助用户信息
     * @param insAppCarouselMedias
     * @param sponsoredUid
     * @param taggedUid
     */
    public void getChildrenMediaTaggerAndSponsored(List<InsAppCarouselMedia> insAppCarouselMedias,Set<Long> sponsoredUid,Set<Long> taggedUid){
        if(insAppCarouselMedias != null && !insAppCarouselMedias.isEmpty()){
            for(InsAppCarouselMedia insAppCarouselMedia : insAppCarouselMedias){
                if(insAppCarouselMedia != null){
                    // 子帖子中的赞助人
                    getSponsored(insAppCarouselMedia.getSponsorTags(),sponsoredUid);
                    // 子帖子中的提及人
                    if(insAppCarouselMedia.getUsertags() != null && insAppCarouselMedia.getUsertags().getIn() != null
                            && !insAppCarouselMedia.getUsertags().getIn().isEmpty()){
                        getUserTagger(insAppCarouselMedia.getUsertags().getIn(),taggedUid);
                    }
                }
            }
        }
    }

    /**
     * 保存提及的用户
     * @param edges
     */
    public void getUserTagger(List<InsAppMediaTaggedBody> edges, Set<Long> uids){
        for(InsAppMediaTaggedBody edge : edges){
            if(edge != null && edge.getUser() != null
                    && edge.getUser().getUid() > 0
                    && !uids.contains(edge.getUser().getUid())){
                uids.add(edge.getUser().getUid());
                sendPriorityCollectBloggerMsg(edge.getUser().getUid(),edge.getUser().getUsername(),
                        edge.getUser().getPicUrl(),edge.getUser().getIsPrivate() != null ? edge.getUser().getIsPrivate() : false,
                        InsBloggerMsgTypeCodeEnum.CLP_SRC_MEDIA_TAGGER.getCode());
            }
        }
    }

    /**
     * 保存提及的用户
     * @param shortcode
     * @param uid
     * @param nowTime
     */
    public void saveUserTagger(String shortcode,long uid,long takenTime, List<Long> taggedUids, long nowTime){
        //初始化
        List<InsRelMediaTaggerUser> insRelMediaTaggerUserList = new ArrayList<>(taggedUids.size());
        for(long taggedUid : taggedUids){
            InsRelMediaTaggerUser insRelMediaTaggerUser = new InsRelMediaTaggerUser();
            insRelMediaTaggerUser.setShortcode(shortcode);
            insRelMediaTaggerUser.setTakenTime(takenTime);
            insRelMediaTaggerUser.setUid(uid);
            insRelMediaTaggerUser.setTaggerUid(taggedUid);
            insRelMediaTaggerUser.setCreateTime(nowTime);
            insRelMediaTaggerUserList.add(insRelMediaTaggerUser);
        }
        //批量保存关系
        if(!insRelMediaTaggerUserList.isEmpty()){
            insRelMediaTaggerUserDao.batchInsert(insRelMediaTaggerUserList);
        }
    }

    /**
     * 保存地理位置
     * @param nowTime
     * @param insMedia
     */
    public void saveLocation(long nowTime,InsMedia insMedia){
        try {
            InsLocation insLocationDba = insLocationDao.findByLocationId(insMedia.getLocationId());

            // 判断地理位置是否保存
            // todo 批量保存
            if (insLocationDba == null) {
                InsLocation insLocation = new InsLocation();
                insLocation.setLocationId(insMedia.getLocationId());
                insLocation.setLocationName(insMedia.getLocationName());
                insLocation.setCreateTime(nowTime);
                insLocationDao.save(insLocation);
            }
        }catch (Exception e){
            log.error("insert media location error, msg: {}",e.getMessage());
        }
    }

    /**
     * 优先采集播主信息
     * @param uid
     * @param userName
     * @param picUrl
     * @param isPrivate
     */
    public void sendPriorityCollectBloggerMsg(long uid,String userName,String picUrl,boolean isPrivate,int srcTypeCode){
        if(uid > 0 && insBloggerDao.checkCollecting(uid,true)){
            //这里需要添加验证
            InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
            insBloggerMessage.setUid(uid);
            insBloggerMessage.setUserName(userName);
            insBloggerMessage.setPicUrl(StringUtils.isEmpty(picUrl) ? "" : picUrl);
            insBloggerMessage.setRanking(isPrivate);
            insBloggerMessage.setRetryLimitTimes(insMediaConfig.getInsBloggerPriorityRetryTimes());
            insBloggerMessage.setRetryTimes(insMediaConfig.getInsBloggerPriorityRetryTimes());
            insBloggerMessage.setTypeCode(srcTypeCode);

            kafkaTemplateInsBlogger.send(insMediaConfig.getInsBloggerPriorityCollectOnlineTopic(),insBloggerMessage);
        }
    }

    /**
     * 获取商品相关数据
     * @param insAppMediaBody
     * @param uid
     * @param products
     * @param relMp
     * @param productIds
     */
    public void getProduct(InsAppMediaBody insAppMediaBody,long uid,List<InsProduct> products,
                           List<InsRelMediaProduct> relMp,Set<Long> productIds){
        String shortcode = insAppMediaBody.getShortcode() != null ? insAppMediaBody.getShortcode() : "";
        // 从帖子产品提及里面来
        if(insAppMediaBody.getProductTags() != null && insAppMediaBody.getProductTags().getIn() != null
                && !insAppMediaBody.getProductTags().getIn().isEmpty()){

            getProduct(insAppMediaBody.getProductTags().getIn(),uid,shortcode,
                    insAppMediaBody.getTakenTime(),products,relMp,productIds,
                    insAppMediaBody.getCommentCount(),insAppMediaBody.getLikeCount(),insAppMediaBody.getEr());
        }
        // 从子帖子的产品提及里来
        if(insAppMediaBody.getCarouselMedia() != null && !insAppMediaBody.getCarouselMedia().isEmpty()){
            for(InsAppCarouselMedia insAppCarouselMedia : insAppMediaBody.getCarouselMedia()){
                if(insAppCarouselMedia != null && insAppCarouselMedia.getProductTags() != null
                        && insAppCarouselMedia.getProductTags().getIn() != null
                        && !insAppCarouselMedia.getProductTags().getIn().isEmpty()){
                    getProduct(insAppCarouselMedia.getProductTags().getIn(),uid,shortcode,
                            insAppMediaBody.getTakenTime(),products,relMp,productIds,
                            insAppMediaBody.getCommentCount(),insAppMediaBody.getLikeCount(),insAppMediaBody.getEr());
                }
            }
        }
    }

    /**
     * 获取商品相关数据
     * @param insAppProductTags
     * @param uid
     * @param shortcode
     * @param mediaTakeTime
     * @param products
     * @param relMp
     * @param productIds
     */
    public void getProduct(List<InsAppProductTag> insAppProductTags,
                           long uid,String shortcode, long mediaTakeTime,List<InsProduct> products,
                           List<InsRelMediaProduct> relMp,Set<Long> productIds,Integer commentCount,Long likeCount,Double er){
        // 用来去重帖子商品关联信息
        Set<Long> relMpProductIds = new HashSet<>(insAppProductTags.size());
        long nowTime = System.currentTimeMillis() / 1000;
        boolean isSelf;
        InsAppProductBody insAppProductBody;
        for(InsAppProductTag insAppProductTag : insAppProductTags){
            try{
                if(insAppProductTag != null && insAppProductTag.getProduct() != null
                        && insAppProductTag.getProduct().getProductId() != null){
                    insAppProductBody = insAppProductTag.getProduct();
                    // 帖子商品关联信息
                    if(!relMpProductIds.contains(insAppProductBody.getProductId())){
                        relMpProductIds.add(insAppProductBody.getProductId());
                        InsRelMediaProduct insRelMediaProduct = new InsRelMediaProduct();
                        insRelMediaProduct.setShortcode(shortcode);
                        insRelMediaProduct.setTakenTime(mediaTakeTime);
                        insRelMediaProduct.setCreateTime(nowTime);
                        insRelMediaProduct.setProductId(insAppProductBody.getProductId());
                        insRelMediaProduct.setCommentCount(commentCount);
                        insRelMediaProduct.setLikeCount(likeCount);
                        insRelMediaProduct.setEr(er);
                        insRelMediaProduct.setUid(uid);
                        isSelf = false;
                        if(insAppProductBody.getMerchant() != null && insAppProductBody.getMerchant().getUid() != null){
                            isSelf = uid == insAppProductTag.getProduct().getMerchant().getUid();
                        }
                        insRelMediaProduct.setIsSelf(isSelf);
                        relMp.add(insRelMediaProduct);
                    }
                    // 商品信息
                    if(!productIds.contains(insAppProductBody.getProductId())){
                        InsProduct insProduct = InsAppProductConversion.mapper(insAppProductBody);
                        insProduct.setLastMediaTagTime(mediaTakeTime);
                        insProduct.setCreateTime(nowTime);
                        insProduct.setLastCollectTime(nowTime);
                        productIds.add(insProduct.getProductId());
                        products.add(insProduct);

                        // 发送优先收入商品商家信息
                        if(uid != insProduct.getMerchantUid() && insAppProductBody.getMerchant() != null){
                            sendPriorityCollectBloggerMsg(uid,insAppProductBody.getMerchant().getUsername(),insAppProductBody.getMerchant().getPicUrl(),
                                    insAppProductBody.getMerchant().getIsPrivate() != null && insAppProductBody.getMerchant().getIsPrivate(),
                                    InsBloggerMsgTypeCodeEnum.CLP_SRC_MERCHANT.getCode());
                        }
                    }
                }
            }catch (Exception e){
                log.error("ins product merge error, msg :{}",e.getMessage());
            }
        }
    }

    /**
     * 发送后续流程消息
     * @param insBloggerMessage
     * @param uid
     * @param insBloggerMain
     * @param start
     */
    public void sendMsg(InsBloggerMessage insBloggerMessage, long uid, InsBloggerMain insBloggerMain, long start){
        // 最近采集时间限制
        long lastCollectTime = start / 1000 - insMediaConfig.getInsCommentIntervalDay() * 24 * 3600;
        try{
            // 发送分析视频分数
            sendAnalysisMediaScoreMsg(insBloggerMessage);
            // 发送采集评论
            if(insBloggerMessage.getTypeCode() != InsBloggerMsgTypeCodeEnum.UID_UPDATE_F_BLOGGER_TIMING_MIDDLE.getCode()
                    && insBloggerMain.getLastCommentCollectTime() < lastCollectTime){
                sendCollectCommentMsg(insBloggerMessage);
            }

            // 发送识别广告帖的消息
            sendIdentifyAdMediaMsg(insBloggerMessage);

            // 发送分析播主分类的消息
            if(StringUtils.isEmpty(insBloggerMain.getOverallCategoryName())){
                sendAnalysisCategoryMsg(insBloggerMessage);
            }
        }catch (Exception e){
            log.error("InsMedia send msg error,uid:{},msg : {}" ,uid,e.getMessage());
        }
    }

    /**
     * 保存商品相关操作，同时调用标记商品接口获取标记商品的帖子
     * @param products
     * @param relMp
     * @param productIds
     */
    public void saveProductsAndDoSpider(List<InsProduct> products,List<InsRelMediaProduct> relMp,Set<Long> productIds){
        List<Long> productIdList = new ArrayList<>(productIds);
        // 保存商品
        saveProduct(products,productIdList);
        // 保存商品和帖子关系
        insProductService.saveRelMp(relMp);
        // 调用标记商品接口获取标记商品的帖子
        sendToSpiderProduct(productIdList);
    }

    /**
     * 保存商品
     * @param products
     * @param productIds
     */
    public void saveProduct(List<InsProduct> products,List<Long> productIds){
        try{
            List<Long> productExist = insProductDao.findExistUid(productIds);
            if(productExist != null && !productExist.isEmpty()){
                List<InsProduct> insertProduct = new ArrayList<>(products.size());
                List<InsProduct> updateProduct = new ArrayList<>(products.size());
                Set<Long> existUid = new HashSet<>(productExist);
                products.forEach(product -> {
                    if(existUid.contains(product.getProductId())){
                        updateProduct.add(product);
                    }else{
                        insertProduct.add(product);
                    }
                });
                insProductDao.insertBatch(insertProduct);
                insProductDao.updateBatch(updateProduct);

                // 发送分析商品分类
                insProductService.sendAnalysisProductCategoryMsg(insertProduct);
            }else{
                insProductDao.insertBatch(products);
                // 发送分析商品分类
                insProductService.sendAnalysisProductCategoryMsg(products);
            }
        }catch (Exception e){
            log.error("save ins media Product error,msg:{}",e.getMessage());
        }

        List<InsProductEsMapping> insProducts = insProductMapping.mapper(insProductDao.findByProductIds(productIds));
        saveToEs(insProducts);
    }

    /**
     * 同步数据到ES
     * @param insProductMappers
     */
    private void saveToEs(List<InsProductEsMapping> insProductMappers){
        RestClientHelper.upsert(insMediaConfig.getProductIndex(),insProductMappers);
    }

    /**
     * 调用标记商品接口获取标记商品的帖子
     * @param productIdList
     */
    public void sendToSpiderProduct(List<Long> productIdList){
        productIdList.forEach( productId -> {
            InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
            insBloggerMessage.setId(productId);
            insBloggerMessage.setUid(productId);
            insBloggerMessage.setRetryLimitTimes(insMediaConfig.getInsAppProductMediaRetryTimes());
            insBloggerMessage.setRetryTimes(insMediaConfig.getInsAppProductMediaRetryTimes());
            kafkaTemplateInsBlogger.send(insMediaConfig.getInsAppProductMediaOnlineTopic(),insBloggerMessage);
        });
    }
}
